package com.wdh.dbutil;

import cn.afterturn.easypoi.excel.ExcelImportUtil;
import cn.afterturn.easypoi.excel.entity.ImportParams;
import cn.hutool.core.io.file.FileReader;
import cn.hutool.core.io.file.FileWriter;
import com.wdh.dbutil.domain.KafkaTopicDto;

import java.io.File;
import java.io.FileInputStream;
import java.util.ArrayList;
import java.util.List;

/**
 * 生成kafka建立topic命令
 * */
public class KafkaTopicCreateUtil {


    /**
     * /opt/kafka_2.11-0.10.2.1/bin/kafka-topics.sh --create --zookeeper zk-cs:2181/kafka --replication-factor  1 --partitions  1 --topic Rc_PUB_Receive_SPAT
     * /opt/kafka_2.11-0.10.2.1/bin/kafka-topics.sh --create --zookeeper zk-cs:2181 --replication-factor  1 --partitions  1 --topic Rc_PUB_Receive_SPAT
     * */
    public static void main(String[] args) throws Exception{
        createTopicV2();
    }

    /**
     * 解析xls文件
     * */
    private static void createTopicV1()throws Exception{
        String kafkaServer = "127.0.0.1:9092";
        File file = new File("d:/regionKafka.xlsx");
        String cmd = "/usr/local/ykzx/kafka/bin/kafka-topics.sh --create --bootstrap-server %s --replication-factor  %d --partitions  %d  --topic %s";
        ImportParams importParams = new ImportParams();
        List<KafkaTopicDto> list = ExcelImportUtil.importExcel(new FileInputStream(file), KafkaTopicDto.class, importParams);
        List<String> cmdList = new ArrayList<>();
        cmdList.add("#!/bin/bash");
        for (KafkaTopicDto dto : list) {
            String topic = dto.getTopic().trim();
            Integer partitionCount = dto.getPartitionCount();
            Integer copyCount = dto.getCopyCount();
            String curCmd = String.format(cmd,kafkaServer,copyCount,partitionCount,topic);
            cmdList.add(curCmd);
        }

        FileWriter writer = new FileWriter("d:/cmdRegion.txt");
        writer.writeLines(cmdList);
    }

    /**
     * 读取txt文件
     * */
    private static void createTopicV2()throws Exception{
        FileReader reader = new FileReader("d:/topic.txt");
        List<String> topiciList = reader.readLines();
        List<String> cmdList = new ArrayList<>();
        cmdList.add("#!/bin/bash");
        String cmd = "/usr/local/ykzx/kafka/bin/kafka-topics.sh --create --bootstrap-server 127.0.0.1:9092 --replication-factor  1 --partitions  3 --topic %s";

        for (String topic : topiciList) {
            String curCmd = String.format(cmd,topic);
            cmdList.add(curCmd);
        }
        FileWriter writer = new FileWriter("d:/createTopic.txt");
        writer.writeLines(cmdList);
    }
}
